package spark

import fun.functions.{getKafkaData, openSpark, sendKafka}

object streamKafka3 {
  def main(args: Array[String]): Unit = {
    val streamingContext = openSpark
    val lines = getKafkaData(streamingContext)

    // 更新状态
    val updateGenderCount = (values: Seq[Int], state: Option[Int]) => {
      val currentCount = values.sum
      val previousCount = state.getOrElse(0)
      Some(currentCount + previousCount)
    }

    /**
     * 在籍和不在籍男女生 -> （在籍，性别） 人数
     * */

    val genderEnrollCount = lines
      .map {
        line =>
          val gender = line.split("\t")(2)
          val enrolled = line.split("\t")(6)
          ((gender, enrolled), 1)
      }
      .updateStateByKey(updateGenderCount)

    genderEnrollCount.foreachRDD { rdd =>
      val genderEnrollCountList = rdd.collect().toList
      val record: StringBuilder = new StringBuilder("{")
      genderEnrollCountList.foreach {
        case (genderEnroll, count) =>
          record.append(s""""$genderEnroll":$count,""")
      }
      record.setLength(record.length - 1)
      record.append("}")
      sendKafka("result2", record)
    }
    streamingContext.start()
    streamingContext.awaitTermination()
  }
}
